package com.yiguo.realtime.app

import com.alibaba.fastjson.JSON
import com.yiguo.gmall.Constansts
import com.yiguo.realtime.beans.OrderInfo
import com.yiguo.realtime.utils.MyKafkaUtil
import org.apache.hadoop.hbase.HBaseConfiguration

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import org.apache.phoenix.spark._
import org.apache.spark.streaming.dstream.DStream

object GMVApp extends BaseApp {
  override var appName: String = "GMVApp"
  override var duration: Int = 10
  def main(args: Array[String]): Unit = {
    run{
      val ds = MyKafkaUtil.getKafkaStream(
        Constansts.GMALL_ORER_INFO, streamingContext)
      //{"payment_way":"1","delivery_address":"berwkIKzvEKNIevOKgEz","consignee":"dYItVy","create_time":"2021-07-07 07:28:04","order_comment":"LhDupqKBEoIABeRXoyEz","expire_time":"","order_status":"1","out_trade_no":"6959055074","tracking_no":"","total_amount":"585.0","user_id":"1","img_url":"",
      // "province_id":"1","consignee_tel":"13531684494","trade_body":"","id":"1","parent_order_id":"","operate_time":""}
      //封装样例类
      val reslutds: DStream[OrderInfo] = ds.map(record => {
        val orderInfo: OrderInfo = JSON.parseObject(record.value(), classOf[OrderInfo])

        val formatter: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
        val formatterdate: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd")

        val creattime: LocalDateTime = LocalDateTime.parse(orderInfo.create_time, formatter)
        orderInfo.create_date = creattime.format(formatterdate)
        orderInfo.create_hour = creattime.getHour.toString
        //订单的明细数据使用脱敏，这里只演示 手机号脱敏
        orderInfo.consignee_tel = orderInfo.consignee_tel.replaceAll("(\\w{3})\\w*(\\w{4})", "$1****$2")
        orderInfo
      })
      //写入Hbase
       reslutds.foreachRDD(rdd=>{
         rdd.saveToPhoenix("GMALL2020_ORDER_INFO",Seq("ID","PROVINCE_ID", "CONSIGNEE", "ORDER_COMMENT", "CONSIGNEE_TEL", "ORDER_STATUS", "PAYMENT_WAY", "USER_ID","IMG_URL", "TOTAL_AMOUNT", "EXPIRE_TIME", "DELIVERY_ADDRESS", "CREATE_TIME","OPERATE_TIME","TRACKING_NO","PARENT_ORDER_ID","OUT_TRADE_NO", "TRADE_BODY", "CREATE_DATE", "CREATE_HOUR"),
           //如果将来项目中如果有Hbse的配置文件的话 new Configuration 配置文件读不了
           HBaseConfiguration.create(),
           //封装zk地址
           Some("hadoop102,hadoop103,hadoop104:2181"))
       })
      println("run 再次运行")
    }
  }
}
